package com.exd.flink.stream.function;

import java.util.Base64;
import java.util.Date;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import com.exceeddata.sdk.vdata.data.VDataMeta;
import com.exceeddata.sdk.vdata.data.VDataReader;
import com.exceeddata.sdk.vdata.data.binary.BinaryLittleEndianUtils;
import com.exd.flink.common.DomainControlFlag;
import com.exd.flink.common.OutputVdataTagBase;
import com.exd.flink.domain.VDataBean;
import com.exd.flink.domain.VDataCountBean;
import com.exd.flink.domain.VDataErrorBean;

/**
 * 描述: 合并数据
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class VdataMergeFunction extends KeyedProcessFunction<String, VDataBean, String> {

	private static final long serialVersionUID = 1L;

	ValueState<VDataCountBean> countVdata;

	MapState<Integer, VDataBean> vdataMergeMap;

	@Override
	public void open(Configuration parameters) throws Exception {
		countVdata = getRuntimeContext().getState(new ValueStateDescriptor<VDataCountBean>("count-data", VDataCountBean.class));
		vdataMergeMap = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, VDataBean>("vdata-merge", Integer.class, VDataBean.class));
	}

	@Override
	public void processElement(VDataBean vdataBean, Context context, Collector<String> collector) throws Exception {
		int sliceNum = vdataBean.getSliceNum();
		int sliceLast = vdataBean.getSliceLast();
		int length = vdataBean.getData().length() - 8;
		VDataCountBean dataCountBean = countVdata.value();
		if (dataCountBean == null)dataCountBean = new VDataCountBean();
		// 判断是否又重复的数据
		VDataBean vDataBeanNullFlag = vdataMergeMap.get(sliceNum);
		if(vDataBeanNullFlag==null) {
			dataCountBean.setNowNum(dataCountBean.getNowNum() + 1);
			vdataMergeMap.put(sliceNum, vdataBean);
			dataCountBean.setByteLengthTotal(dataCountBean.getByteLengthTotal() + length);
		}
		if (sliceLast == 1) {
			dataCountBean.setTotalNum(sliceNum + 1);
		}
		countVdata.update(dataCountBean);
		context.timerService().registerProcessingTimeTimer(3000);
	}

	@Override
	public void onTimer(long timestamp, OnTimerContext context, Collector<String> collector) throws Exception {
		// 判断切片是否完整
		VDataCountBean dataCountBean = countVdata.value();
		if (dataCountBean.getNowNum() == dataCountBean.getTotalNum()) {
			try {
				String service_id = null;
				byte[] buffer = new byte[dataCountBean.getByteLengthTotal()];
				for (int i = 0, p = 0; i < dataCountBean.getNowNum(); i++) {
					VDataBean vDataBean = vdataMergeMap.get(i);
					service_id = vDataBean.getServiceId();
					byte[] packetWithMetadata = Base64.getDecoder().decode(vDataBean.getData());
					System.arraycopy(packetWithMetadata, 8, buffer, p, packetWithMetadata.length - 8);
					p += packetWithMetadata.length - 8;
				}
				// 解析合并完成的数据是否正确
				VDataMeta vdataMeta = VDataReader.getMeta(buffer);
				if(vdataMeta == null || vdataMeta.getExtendedInfo() == null || vdataMeta.getExtendedInfo().length ==0) {
					throw new RuntimeException("VdataMeta or ExtendedInfo IS NULL");
				}
		        long finalRolloutId = -1L;
		        long trigger_start_time_long = -1L;
		        long trigger_end_time_long = -1L;
		        if (vdataMeta.getFormatVersion() >= 23) {
		        	if (vdataMeta.getExtendedInfo().length < 8) {
		        		throw new RuntimeException("ExtendedInfo小于8");
		        	}
		            finalRolloutId = BinaryLittleEndianUtils.bytesToINT64(vdataMeta.getExtendedInfo(), 0);
		            trigger_start_time_long = BinaryLittleEndianUtils.bytesToINT64(vdataMeta.getExtendedInfo(), 8);
		            trigger_end_time_long = BinaryLittleEndianUtils.bytesToINT64(vdataMeta.getExtendedInfo(), 16);
		            if (finalRolloutId < 0) {
		            	throw new RuntimeException("RolloutId小于0");
		            }
		        }else if (vdataMeta.getFormatVersion() == 22) {
		        	if (vdataMeta.getExtendedInfo().length < 24) {
		        		throw new RuntimeException("ExtendedInfo小于24");
		        	}else {
		        		String softwareId = BinaryLittleEndianUtils.bytesToANSI(vdataMeta.getExtendedInfo(), 0, 24);
		        		 if (softwareId.equals("606ac2a697709628256fe2c") || softwareId.equals("606ac2a697709628256fe2c0")) {
		        			 if (service_id.equals("8210")) {
		        				 finalRolloutId = 17;
		        			 } else if (service_id.equals("4124")) {
		        				 finalRolloutId = 26;
		        			 }else {
		        				 finalRolloutId = 26;
		        			 }
		        		 }
		        	}
	                trigger_start_time_long = BinaryLittleEndianUtils.bytesToINT64(vdataMeta.getExtendedInfo(), 24);
	                trigger_end_time_long = BinaryLittleEndianUtils.bytesToINT64(vdataMeta.getExtendedInfo(), 32);
		        }else {
		        	if (vdataMeta.getExtendedInfo().length < 24) {
		        		throw new RuntimeException("ExtendedInfo小于24");
		        	}
		        }
		        // 封装合并完成的数据
		        VDataBean vDataBean = vdataMergeMap.get(0);
		        vDataBean.setDataByte(buffer);
		        vDataBean.setDeployId(finalRolloutId);
		        vDataBean.setTriggerStartTime(trigger_start_time_long);
		        vDataBean.setTriggerEndTime(trigger_end_time_long);
		        // 获取域控标识
		        String serviceId = vDataBean.getServiceId();
		        if (finalRolloutId == -1 || finalRolloutId == 17 || finalRolloutId == 26) {
		        	// 全量数据
					if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
						context.output(OutputVdataTagBase.BDCM_VDATA_PACKAGE_FULL, vDataBean);
						// 索引数据，不需要data数据，减小数据的大小
						vDataBean.setDataByte(null);
						vDataBean.setData(null);
						vDataBean.setServiceName("BDCM");
						context.output(OutputVdataTagBase.BDCM_VDATA_PACKAGE_FULL_INDEX, vDataBean);
					} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
						context.output(OutputVdataTagBase.VDCM_VDATA_PACKAGE_FULL, vDataBean);
						// 索引数据，不需要data数据，减小数据的大小
						vDataBean.setDataByte(null);
						vDataBean.setData(null);
						vDataBean.setServiceName("VDCM");
						context.output(OutputVdataTagBase.VDCM_VDATA_PACKAGE_FULL_INDEX, vDataBean);
					}
		        } else if (String.valueOf(finalRolloutId).startsWith("2")) {
		        	// 数采数据
					if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
						context.output(OutputVdataTagBase.BDCM_VDATA_PACKAGE_MINI, vDataBean);
					} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
						context.output(OutputVdataTagBase.VDCM_VDATA_PACKAGE_MINI, vDataBean);
					}
		        } else if (String.valueOf(finalRolloutId).startsWith("3")) {
		        	// 事件数据
					if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
						context.output(OutputVdataTagBase.BDCM_VDATA_PACKAGE_EVENT, vDataBean);
					} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
						context.output(OutputVdataTagBase.VDCM_VDATA_PACKAGE_EVENT, vDataBean);
					}
		        } else {
		        	throw new RuntimeException("RolloutId IS NOT IN -1 17 26");
		        }
			} catch (Exception exception) {
				// 保存合并异常的数据
				vdataMergeMap.values().forEach(vDataBean ->{
					String serviceId = vDataBean.getServiceId();
					VDataErrorBean vDataErrorBean = new VDataErrorBean();
					vDataErrorBean.setServiceId(serviceId);
					vDataErrorBean.setVin(vDataBean.getVin());
					vDataErrorBean.setYear(vDataBean.getYear());
					vDataErrorBean.setMonth(vDataBean.getMonth());
					vDataErrorBean.setDay(vDataBean.getDay());
					vDataErrorBean.setHour(vDataBean.getHour());
					vDataErrorBean.setMin(vDataBean.getMin());
					vDataErrorBean.setSec(vDataBean.getSec());
					vDataErrorBean.setMethodId(vDataBean.getMethodId());
					vDataErrorBean.setMajorVersion(vDataBean.getMajorVersion());
					vDataErrorBean.setSamplingTime(vDataBean.getSamplingTime());
					vDataErrorBean.setTboxSendTimestamp(vDataBean.getTboxSendTimestamp());
					vDataErrorBean.setReceiveTimestamp(vDataBean.getReceiveTimestamp());
					vDataErrorBean.setData(vDataBean.getData());
					vDataErrorBean.setErrorType(0);
					vDataErrorBean.setErrorDescribe("整包数据解析异常");
					vDataErrorBean.setErrorMessage(exception.getMessage());
					vDataErrorBean.setCreateTime(new Date());
					// 输出异常的数据,根据不同的域控，输出到不同的流
					if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
						context.output(OutputVdataTagBase.BDCM_ANALYTICAL_VDATA_ERROR, vDataErrorBean);
					} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
						context.output(OutputVdataTagBase.VDCM_ANALYTICAL_VDATA_ERROR, vDataErrorBean);
					}	
				});
			}
		} else {
			// 保存不完整的切片，进行定时任务合并
			vdataMergeMap.values().forEach(vDataBean ->{
				String serviceId = vDataBean.getServiceId();
				if (serviceId.equalsIgnoreCase(DomainControlFlag.BDCM_VDATA)) {
					context.output(OutputVdataTagBase.BDCM_VDATA_SHARD, vDataBean);
				} else if (serviceId.equalsIgnoreCase(DomainControlFlag.VDCM_VDATA)) {
					context.output(OutputVdataTagBase.VDCM_VDATA_SHARD, vDataBean);
				}
			});
		}
	}

}
